package org.example.config.kafka;

import com.google.common.base.Strings;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Random;

/**
 * @version: 1.00.00
 * @description: 分区策略
 * @copyright: Copyright (c) 2023 立林科技 All Rights Reserved
 * @company: 厦门立林科技有限公司
 * @date: 2023-12-11 15:33
 */
public class myPartitionStategy implements Partitioner {
    private Random random;
    @Override
    public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
        List<PartitionInfo> partitionInfoList = cluster.availablePartitionsForTopic(s);
        String key = (String) o;
        return Strings.isNullOrEmpty(key) ? random.nextInt(partitionInfoList.size()-1) : Objects.hash(key) % partitionInfoList.size();
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> map) {
        random = new Random();
    }
}
